/*
 * LANES.C   	                          Copyright (c) 2007-08, Asko Kauppi
 *
 * Multithreading in Lua.
 *
 * History:
 *      24-Jun-08 .. 14-Jul-08 AKa: Major revise, Lanes 2008 version
 *          ...
 *      18-Sep-06 AKa: Started the module.
 *
 * Platforms (tested per release):
 *      OS X (10.5.4 PowerPC, Intel)
 *      Linux (Ubuntu x86, NSLU2 ARM, AMD-64)
 *      Win32 (Windows XP Home SP2)
 *      FreeBSD (6.2-RELEASE i386)
 *
 * Platforms (tbd):
 *      Win64 - testers appreciated
 *      QNX - shouldn't be hard, if someone volunteers
 *      Sun Solaris - volunteers appreciated
 *
 * References:
 *      "Porting multithreaded applications from Win32 to Mac OS X":
 *      <http://developer.apple.com/macosx/multithreadedprogramming.html>
 *
 *      Pthreads:
 *      <http://vergil.chemistry.gatech.edu/resources/programming/threads.html>
 *
 *      MSDN: <http://msdn2.microsoft.com/en-us/library/ms686679.aspx>
 *
 *      <http://ridiculousfish.com/blog/archives/2007/02/17/barrier>
 *
 * Defines:
 *      -DLINUX_SCHED_RR: all threads are lifted to SCHED_RR category, to
 *          allow negative priorities (-2,-1) be used. Even without this,
 *          using priorities will require 'sudo' privileges on Linux.
 *
 *      -D_GNU_SOURCE required on Linux, to bring 'pthread_yield()' about.
 *
 *      -DTBB_ALLOC uses Intel Threading Building Blocks scalable allocator
 *      <http://threadingbuildingblocks.org>
 *
 * Design notes:
 *
 *      Waiting for multiple lanes to get ready (or messages from them) is
 *      implemented using FIFOs, not OS features. PThreads lacks a wait-for-
 *      multiple concept altogether, and Win32 API's WaitForMultipleObjects()
 *      has a ridicilous limit of 64 objects.
 *
 *      The hilarious lack of 'pthread_timedjoin()' from the Pthread official
 *      API caused to avoid a 'pthread_join()' altogether. Funny "standard",
 *      but this approach has many benefits to it, s.a. reusing a thread id
 *      immediately when the thread's done vs. when the main thread would have
 *      time to 'join' with it. Better this way.
 *
 * Bugs:
 *
 *      The "self" recursive parallel call mechanism does not work (currently
 *      disabled by #if 0)
 *
 * To-do:
 *
 *      ...
 */

const char *VERSION= "20080714";

/*
===============================================================================

Copyright (C) 2007-08 Asko Kauppi <akauppi@gmail.com>

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in
all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.  IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
THE SOFTWARE.

===============================================================================
*/
#include <string.h>
#include <stdio.h>
#include <ctype.h>

#include "lua.h"
#include "lauxlib.h"

#include "threading.h"
#include "tools.h"

/* geteuid() */
#ifdef PLATFORM_LINUX
# include <unistd.h>
# include <sys/types.h>
#endif

#define KEEPER_STATES_N 6

/* Time to wait for a thread to detect a cancellation request, before
* forcefully killing it.
*/
#define CANCEL_WAIT_MS (100)

/*
* Lua code for the keeper states (baked in)
*/
char keeper_chunk[]=
#include "keeper.lch"

    /*
    * Lua code for the timer state (baked in)
    */
    char timer_chunk[]=
#include "timer.lch"

        static LOCK_T require_cs;

static void cancel_test( lua_State *L );
#define CANCEL_TEST_KEY ((void*)cancel_test)    // used as registry key

struct s_Linda;


/*
* Panic function for hidden states, gives line number info.
*
* Just for debugging, there should be no errors in hidden state code.
*/
static int panic( lua_State *L )
{
    const char *err= lua_tostring(L,-1);
    const char *trace;

    lua_atpanic( L, NULL );

    STACK_GROW(L,1);

    luaL_where(L, 0 /*current level*/);   // pushes a call stack
    trace= lua_tostring(L,-1);

    fprintf( stderr, "HIDDEN STATE: %s %s\n", err, trace ? trace : "(no trace)" );
    return 0;
}


/*---=== Keeper states ===---
*/

/*
* Pool of keeper states
*
* Access to keeper states is locked (only one OS thread at a time) so the
* bigger the pool, the less chances of unnecessary waits. Lindas map to the
* keepers randomly, by a hash.
*/
struct s_Keeper
{
    LOCK_T lock;
    lua_State *L;
} keeper[KEEPER_STATES_N];

static bool_t init_keepers(void)
{
    unsigned int i;
    for ( i=0; i<KEEPER_STATES_N; i++ )
    {

        // Initialize Keeper states with bare minimum of libs (those required
        // by 'keeper.lua')
        //
        lua_State *L= luaL_newstate();
        if (!L) return FALSE;

        luaG_openlibs( L, "io,table" );     // 'io' for debugging messages

        // Read in the preloaded chunk (and run it)
        //
        if (luaL_loadbuffer( L, keeper_chunk, sizeof(keeper_chunk), "=keeper" ))
            return FALSE;   // LUA_ERRMEM

        if (lua_pcall( L, 0 /*args*/, 0 /*results*/, 0 /*errfunc*/ ))
            return FALSE;   // LUA_ERRRUN / LUA_ERRMEM / LUA_ERRERR

        lua_atpanic( L, panic );

        LOCK_INIT( &keeper[i].lock );
        keeper[i].L= L;
    }
    return TRUE;
}

static
struct s_Keeper *keeper_acquire( const void *ptr )
{

    /*
    * Any hashing will do that maps pointers to 0..KEEPER_STATES_N-1
    * consistently.
    *
    * Pointers are often aligned by 8 or so - ignore the low order bits
    */
    unsigned int i= ((unsigned long)(ptr) >> 3) % KEEPER_STATES_N;
    struct s_Keeper *K= &keeper[i];

    LOCK_START( &K->lock );
    return K;
}

static
void keeper_release( struct s_Keeper *K )
{
    LOCK_END( &K->lock );
}

/*
* Call a function ('func_name') in the keeper state, and pass on the returned
* values to 'L'.
*
* 'linda':          deep Linda pointer (used only as a unique table key, first parameter)
* 'starting_index': first of the rest of parameters (none if 0)
*
* Returns:  number of return values (pushed to 'L')
*/
static
int keeper_call( lua_State* K, const char *func_name,
                 lua_State *L, struct s_Linda *linda, uint_t starting_index )
{

    int args= starting_index ? (lua_gettop(L) - starting_index +1) : 0;
    int Ktos= lua_gettop(K);
    int retvals;

    lua_getglobal( K, func_name );
    ASSERT_L( lua_isfunction(K,-1) );

    STACK_GROW( K, 1 );
    lua_pushlightuserdata( K, linda );

    luaG_inter_copy( L,K, args );   // L->K
    lua_call( K, 1+args, LUA_MULTRET );

    retvals= lua_gettop(K) - Ktos;

    luaG_inter_move( K,L, retvals );    // K->L
    return retvals;
}


/*---=== Linda ===---
*/

/*
* To allow same Linda be used from multiple Lua states, the actual userdata
* must be only a pointer. This should work... *knock*knock*
*
* Actual data is kept within a keeper state, which is hashed by the 's_Linda'
* pointer (which is same to all userdatas pointing to it).
*/
struct s_Linda
{
    SIGNAL_T read_happened;
    SIGNAL_T write_happened;

    volatile uint_t refcount;   // number of copies referring to this
    // inc/dec'd under keeper state's lock influence
};

static int LG_linda_send( lua_State *L );
static int LG_linda_receive( lua_State *L );
static int LG_linda_set( lua_State *L );
static int LG_linda_get( lua_State *L );

#define lua_toLinda(L,n) ((struct s_Linda *)luaG_toshared( L, LG_linda_id, n ))

/*
* tbl= linda_id()
* void= linda_id( lightuserdata )
*
* Identity function of a shared userdata object.
*
* If called without parameters, returns a metatable for the objects ('__gc'
* method not needed; will be handled by 'luaG_...').
*
* If called with a light userdata object, cleans up the contents.
*/
LUAG_FUNC( linda_id )
{

    struct s_Linda *s= lua_touserdata(L,1);
    if (s)
    {
        /*
        * Last user of this Linda has gc'ed; clean the structure and associated
        * data in the keeper's.
        */
        struct s_Keeper *K= keeper_acquire(s);
        {
            keeper_call( K->L, "clear", L, s, 0 );
        }
        keeper_release(K);

        SIGNAL_FREE( &s->read_happened );
        SIGNAL_FREE( &s->write_happened );

        return 0;
    }

    STACK_CHECK(L)
    lua_newtable(L);

    lua_newtable( L );
    STACK_MID(2)
    //
    // [-2]: linda metatable
    // [-1]: metatable's to-be .__index table

    lua_pushcfunction( L, LG_linda_send );
    lua_setfield( L, -2, "send" );
    STACK_MID(2)

    lua_pushcfunction( L, LG_linda_receive );
    lua_setfield( L, -2, "receive" );
    STACK_MID(2)

    lua_pushcfunction( L, LG_linda_set );
    lua_setfield( L, -2, "set" );
    STACK_MID(2)

    lua_pushcfunction( L, LG_linda_get );
    lua_setfield( L, -2, "get" );
    STACK_MID(2)

    lua_setfield( L, -2, "__index" );
    STACK_END(1)

    return 1;
}


/*
* linda_ud= linda_new()
*/
LUAG_FUNC( linda_new )
{

    /* The returned struct is allocated separately of Lua stack; we might no
     * longer be around when last reference to it is being released. The userdata
     * proxy is pushed to Lua stack.
     */
    struct s_Linda *s= (struct s_Linda *) luaG_newshared( L, LG_linda_id, sizeof(struct s_Linda) );
    ASSERT_L(s);

    /* Initialize the shared structure
    */
    SIGNAL_INIT( &s->read_happened );
    SIGNAL_INIT( &s->write_happened );

    return 1;
}


/*
* bool= linda_send( linda_ud, key_num|str|bool, value [, ...] [,timeout_secs_num] )
*
* Send a value to Linda.
*
* If 'timeout_secs' is given (and >= 0.0), set the value if the 'slot' is free,
* but otherwise wait, or timeout if it does not get free.
*
* If last param is not there (default), add the value to a queue.
*
* Returns:  'true' if the value was set/queued
*           'false' for timeout
*/
LUAG_FUNC( linda_send )
{
    struct s_Linda *linda= lua_toLinda( L, 1 );
    int64_t wait_until=0;
    bool_t timeout= FALSE;

    if ((lua_gettop(L)>=4) && (lua_type(L,-1) == LUA_TNUMBER))
    {
        double secs= lua_tonumber( L,-1 );
        wait_until= (secs<0.0) ? -1 : now_ms() + (long)(secs*1000.0);

        // We can simply pop the timeout away from parameters; 'keeper_call()'
        // will see the rest.
        //
        lua_pop(L,1);
    }

    STACK_GROW(L,1);

    struct s_Keeper *K= keeper_acquire( linda );
    {
        STACK_CHECK(K->L)

        if (wait_until != 0)
        {
            while (TRUE)
            {
                int pushed;
                bool_t empty;

                // Timeout send: only proceed when the previous data is
                // fully consumed
                //
                STACK_GROW( L,1 );
                STACK_MID(0)
                lua_pushvalue( L,2 );

                pushed= keeper_call( K->L, "get", L, linda, lua_gettop(L) );

                ASSERT_L( pushed==1 );  // value/nil
                empty= lua_isnil(L,-1);
                lua_pop(L,2);   // retval + two pushed parameters
                STACK_MID(0)

                if (!empty)
                {
                    long still= -1;
                    if (wait_until>0)
                    {
                        still= (long)( wait_until - now_ms() );
                        if (still<=0)
                        {
                            timeout=TRUE;    // timeout
                            break;
                        }
                    }

                    cancel_test( L );   // testing here causes no delays

                    // K lock will be released for the duration of wait and re-acquired
                    //
                    SIGNAL_WAIT( &linda->read_happened, &K->lock, still );
                }
                STACK_MID(0)
            }
        }

        if (!timeout)
        {
            int pushed= keeper_call( K->L, "send", L, linda, 2 );
            ASSERT_L( pushed==0 );
            SIGNAL_SET( &linda->write_happened );
        }
        STACK_END(0)
    }
    keeper_release(K);

    lua_pushboolean( L, !timeout );
    return 1;
}


/*
* [val [,key]]= linda_receive( linda_ud, key_num|str|bool [, ...] [, timeout_secs_num] )
*
* Receive a value from Linda, consuming it.
*
* Returns:  value received (which is consumed from the slot)
*           key which had it (if multiple keys waited upon)
*/
LUAG_FUNC( linda_receive )
{
    struct s_Linda *linda= lua_toLinda( L, 1 );
    int64_t wait_until= 0;
    int pushed;
    struct s_Keeper *K;

    if (lua_isnumber(L,-1))
    {
        wait_until= now_ms() + (long)(lua_tonumber(L,-1) * 1000.0);
        lua_pop(L,1);   // take timeout away from the parameters
    }

    K= keeper_acquire( linda );
    {
        while (TRUE)
        {
            pushed= keeper_call( K->L, "receive", L, linda, 2 );
            if (pushed)
            {
                // To be done from within the 'K' locking area
                //
                SIGNAL_SET( &linda->read_happened );
                break;

            }
            else
            {
                // Nothing received; wait
                //
                long still= -1;
                if (wait_until)
                {
                    still= (long)( wait_until - now_ms() );
                    if (still<=0) break;  // timeout
                }

                cancel_test( L );   // testing here causes no delays

                // Release the K lock for the duration of wait, and re-acquire
                //
                SIGNAL_WAIT( &linda->write_happened, &K->lock, still );
            }
        }
    }
    keeper_release(K);

    return pushed;
}


/*
* void= linda_set( linda_ud, key_num|str|bool [,value] )
*
* Set a value to Linda.
*
* Existing slot value is replaced, and possible queue entries removed.
*/
LUAG_FUNC( linda_set )
{
    struct s_Linda *linda= lua_toLinda( L, 1 );

    struct s_Keeper *K= keeper_acquire( linda );
    {
        int pushed= keeper_call( K->L, "set", L, linda, 2 );
        ASSERT_L( pushed==0 );
    }
    keeper_release(K);

    if (!lua_isnil(L,3))
    {
        SIGNAL_SET( &linda->write_happened );
    }

    return 0;
}


/*
* [val]= linda_get( linda_ud, key_num|str|bool )
*
* Get a value from Linda.
*/
LUAG_FUNC( linda_get )
{
    struct s_Linda *linda= lua_toLinda( L, 1 );
    int pushed;

    struct s_Keeper *K= keeper_acquire( linda );
    {
        pushed= keeper_call( K->L, "get", L, linda, 2 );
        ASSERT_L( pushed==0 || pushed==1 );
    }
    keeper_release(K);

    return pushed;
}


/*---=== Serialize 'require' calls ===---*/

// TBD: Could possibly do the 'require' serialization at Lua level, using
//      Lindas for locking.

// We serialize 'require' calls, to allow module authors to use simple 'static'
// data structures for initialization constructs. Otherwise, they should
// basically have a critical section within their code. This just makes more
// modules multithreading-ready than would otherwise be.

//---
// ...= require_safe( ... )
//
// The original 'require' function is as our upvalue.
//
static
int require_safe( lua_State *L )
{
    int args= lua_gettop(L);    // [1..args]= parameters

    int idx= lua_upvalueindex(1);
    ASSERT_L( lua_isfunction(L,idx) );

    // Push the function ahead of arguments
    //
    lua_pushvalue( L, idx );
    lua_insert( L, 1 );

    // [1]: function to call
    // [2..args+1]: parameters

    LOCK_START( &require_cs );
    lua_call( L, args, LUA_MULTRET );
    LOCK_END( &require_cs );

    return lua_gettop(L);    // return values
}

static
void serialize_require( lua_State *L )
{
    STACK_CHECK(L)

    // Get the existing function
    //
    lua_getglobal( L, "require" );
    //
    // [-1]: function or nil

    if (lua_isfunction( L,-1) )
    {
        lua_pushcclosure( L, require_safe, 1 /*upvalues*/ );
        //
        // [-1]: closure (serialized)

        lua_setglobal( L, "require" );
    }
    else
    {
        lua_pop( L, 1 );
    }
    STACK_END(0)
}


/*---=== Threads ===---
*/
#undef ERROR
enum e_state
{
    PENDING=1,
    RUNNING=2,
    WAITING=3,
    DONE=4,
    ERROR=5,
    CANCELLED=6
};

// NOTE: values to be changed by either thread, during execution, without
//       locking, are marked "volatile"
//
struct s_lane
{
    THREAD_T thread;
    //
    // M: sub-thread OS thread
    // S: not used

    lua_State *L;
    //
    // M: prepares the state, and reads results
    // S: while S is running, M must keep out of modifying the state

    volatile enum e_state state;
    //
    // M: sets to PENDING (before launching)
    // S: updates -> RUNNING/WAITING -> DONE/ERROR/CANCELLED

    volatile bool_t cancel_request;
    //
    // M: sets to FALSE, flags TRUE for cancel request
    // S: reads to see if cancel is requested

    SIGNAL_T cancel_caught;
    //
    // M: Waited upon at cancellation to know instantly when S has caught
    //    the request
    // S: sets the signal once cancellation is noticed (avoids a kill)

#if (!( (defined PLATFORM_WIN32) || (defined PLATFORM_WINCE)))
    pthread_mutex_t artificial_mutex;
#endif
};

/*
* Check if the thread in question ('L') has been signalled for cancel
* (and do it, if yes).
*
* Called by cancellation hooks and/or pending Linda operations (because then
* the check won't affect performance).
*/
static void cancel_test( lua_State *L )
{
    struct s_lane *s;

    STACK_GROW(L,1);

    STACK_CHECK(L)
    lua_pushlightuserdata( L, CANCEL_TEST_KEY );
    lua_rawget( L, LUA_REGISTRYINDEX );
    s= lua_touserdata( L, -1 );     // userdata/nil
    lua_pop(L,1);
    STACK_END(0)

    // 's' is NULL for the original main state (no-one can cancel that)
    //
    if (s && s->cancel_request)
    {
        // M will clean the state once it sees we're cancelled

        s->state= CANCELLED;
        SIGNAL_SET( &s->cancel_caught );     // wake up master that we did get here

        THREAD_EXIT( &s->thread );
    }
}

static void cancel_hook( lua_State *L, lua_Debug *ar )
{
    (void)ar;
    cancel_test(L);     // if cancelled, does not return
}


//---
// void= _single( [cores_uint=1] )
//
// Limits the process to use only 'cores' CPU cores. To be used for performance
// testing on multicore devices. DEBUGGING ONLY!
//
LUAG_FUNC( _single )
{
    uint_t cores= luaG_optunsigned(L,1,1);

#ifdef PLATFORM_OSX
#ifdef _UTILBINDTHREADTOCPU
    if (cores > 1) luaL_error( L, "Limiting to N>1 cores not possible." );
    // requires 'chudInitialize()'
    utilBindThreadToCPU(0);     // # of CPU to run on (we cannot limit to 2..N CPUs?)
#else
    luaL_error( L, "Not available: compile with _UTILBINDTHREADTOCPU" );
#endif
#else
    luaL_error( L, "not implemented!" );
#endif
    (void)cores;

    return 0;
}


//---
// void= _yield()
//
// Gives execution to another thread. This is NOT required in applications,
// but it helps in test scripts.
//
LUAG_FUNC( _yield )
{
    THREAD_YIELD();
    return 0;
}


/*
* str= lane_error( error_val|str )
*
* Called if there's an error in some lane; add call stack to error message
* just like 'lua.c' normally does.
*
* ".. will be called with the error message and its return value will be the
*     message returned on the stack by lua_pcall."
*/
static int lane_error( lua_State *L )
{
    lua_Debug ar;

    assert( lua_gettop(L)==1 );

    // [1]: plain error message

    if (lua_type(L,1) == LUA_TSTRING)
    {
        fprintf( stderr, "lane_error: %s\n", lua_tostring(L,-1) );
        if (lua_getstack(L, 1 /*level*/, &ar))
        {
            lua_getinfo(L, "Sl", &ar);
            if (ar.currentline > 0)
            {
                STACK_GROW(L,1);
                lua_pushfstring( L, "%s:%d: ", ar.short_src, ar.currentline );
                lua_concat( L, 2 );
            }
        }
    }
    return 1;
}


//---
#if (defined PLATFORM_WIN32) || (defined PLATFORM_POCKETPC)
static THREAD_RETURN_T __stdcall lane_main( void *vs )
#else
static THREAD_RETURN_T lane_main( void *vs )
#endif
{
    struct s_lane *s= (struct s_lane *)vs;
    int rc;

    s->state= RUNNING;  // PENDING -> RUNNING

    STACK_GROW( s->L, 1 );
    lua_pushcfunction( s->L, lane_error );
    lua_insert( s->L, 1 );

    // [1]: error handler
    // [2]: function to run
    // [3..top]: parameters
    //
    rc= lua_pcall( s->L, lua_gettop(s->L)-2, LUA_MULTRET, 1 /*error handler*/ );
    //
    // 0: no error
    // LUA_ERRRUN: a runtime error
    // LUA_ERRMEM: memory allocation error
    // LUA_ERRERR: error while running the error handler (if any)

    lua_remove(s->L,1);    // remove error handler

    // leave results (1..top) or error message (top) on the stack - master will copy them

    s->state= rc==0 ? DONE : ERROR;

    THREAD_EXIT( &s->thread );

    return 0;   // never
}


//---
// lane_ud= thread_new( function, [libs_str],
//                          [cancelstep_uint=0],
//                          [prio_int=0],
//                          [globals_tbl],
//                          [... args ...] )
//
LUAG_FUNC( thread_new )
{
    lua_State *L2;
    struct s_lane *s;

    const char *libs= lua_tostring( L, 2 );
    uint_t cs= luaG_optunsigned( L, 3,0);
    int prio= luaL_optinteger( L, 4,0);
    uint_t glob= luaG_isany(L,5) ? 5:0;

#define FIXED_ARGS (5)
    uint_t args= lua_gettop(L) - FIXED_ARGS;

    if (prio < THREAD_PRIO_MIN || prio > THREAD_PRIO_MAX)
    {
        luaL_error( L, "Priority out of range: %d..+%d (%d)",
                    THREAD_PRIO_MIN, THREAD_PRIO_MAX, prio );
    }

    /* --- Create and prepare the sub state --- */

    L2 = luaL_newstate();   // uses standard 'realloc()'-based allocator,
    // sets the panic callback

    if (!L2) luaL_error( L, "'luaL_newstate()' failed; out of memory" );

    STACK_GROW( L,1 );

    // Setting the globals table (needs to be done before loading stdlibs,
    // and the lane function)
    //
    if (glob!=0)
    {
        STACK_CHECK(L2)
        uint_t L_tos= lua_gettop(L);

        if (!lua_istable(L,glob))
            luaL_error( L, "Expected table, got %s", luaG_typename(L,glob) );

        lua_pushvalue( L, glob );
        luaG_inter_move( L,L2, 1 );     // moves the table to L2

        // L2 [-1]: table of globals

        // "You can change the global environment of a Lua thread using lua_replace"
        // (refman-5.0.pdf p. 30)
        //
        lua_replace( L2, LUA_GLOBALSINDEX );

        lua_settop( L, L_tos );     // eat up temporaries we did (if any left)
        STACK_END(0)
    }

    // Selected libraries
    //
    if (libs)
    {
        luaG_openlibs( L2, libs );

        // base stuff ("print", "error", "assert", ...) are there now
        //
#if 0
        lua_getglobal( L2, "assert" );
        ASSERT_L( lua_iscfunction(L2,-1) );
        lua_pop( L2, 1 );
#endif
    }

    // Lane main function
    //
    STACK_CHECK(L)
    lua_pushvalue( L, 1 );
    luaG_inter_move( L,L2, 1 );    // L->L2
    STACK_END(0)

    ASSERT_L( lua_gettop(L2) == 1 );
    ASSERT_L( lua_isfunction(L2,1) );

    // revive arguments
    //
    if (args) luaG_inter_move( L,L2, args );    // L->L2

    ASSERT_L( lua_gettop(L2) == 1+args );
    ASSERT_L( lua_isfunction(L2,1) );
    ASSERT_L( lua_gettop(L) == FIXED_ARGS );

    // args removed; globals table is now topmost

    // Ready to push the userdata (now that 'args' is used)
    //
    s= lua_newuserdata( L, sizeof(struct s_lane) );
    ASSERT_L(s);
    //memset( s, 0, sizeof(struct s_lane) );

    s->L= L2;
    s->state= PENDING;
    s->cancel_request= FALSE;
    SIGNAL_INIT( &s->cancel_caught );

    // 'artificial_mutex' is a hack for pleasing PThread; we don't have a lock
    // for the 's_lane' structure; nor would we need one.
    //
#if (!( (defined PLATFORM_WIN32) || (defined PLATFORM_WINCE)))
    LOCK_INIT( &s->artificial_mutex );
    LOCK_START( &s->artificial_mutex );
#endif

    // Make calls to 'require' (if enabled for the lane) serialized
    //
    serialize_require( L2 );

    // Place 's' to registry, for 'cancel_test()' (even if 'cs'==0 we still
    // do cancel tests at pending send/receive).
    //
    lua_pushlightuserdata( L2, CANCEL_TEST_KEY );
    lua_pushlightuserdata( L2, s );
    lua_rawset( L2, LUA_REGISTRYINDEX );

    if (cs)
    {
        lua_sethook( L2, cancel_hook, LUA_MASKCOUNT, cs );
    }

    THREAD_CREATE( &s->thread, lane_main, s, prio );

    return 1;
}


//---
// void= thread_gc( lane_ud )
//
// Cleanup for a thread userdata. If the thread is still executing, try to
// cancel it (eventually kill).
//
LUAG_FUNC( thread_gc )
{
    struct s_lane *s= lua_touserdata(L,1);
    int st= s->state;   // read once (volatile)

    if (st!=DONE && st!=ERROR && st!=CANCELLED)
    {

        // GC'ing any still active threads will send them a cancellation request
        // first - this way we might be able to do a nice exit (cleaning up
        // their Lua state). If no, we'll kill them.
        //
        s->cancel_request= TRUE;

        DEBUG()
#if (!( (defined PLATFORM_WIN32) || (defined PLATFORM_WINCE)))
        if (!SIGNAL_WAIT( &s->cancel_caught, &s->artificial_mutex, CANCEL_WAIT_MS ))
#else
        if (!SIGNAL_WAIT( &s->cancel_caught,  CANCEL_WAIT_MS ))
#endif
        {

            // Thread does not react to cancellation request - making a kill
            //
#if 1
            fprintf( stderr, "Killing a thread! (didn't react in %d ms)\n", CANCEL_WAIT_MS );
#endif
            THREAD_KILL( &s->thread );
            THREAD_FREE( &s->thread );
        }
        DEBUG()
    }

    return 0;
}


//---
// void= thread_cancel( lane_ud )
//
// The originator thread asking us specifically to cancel the other thread.
//
LUAG_FUNC( thread_cancel )
{
    struct s_lane *s= lua_touserdata(L,1);
    s->cancel_request= TRUE;    // all we really can do!

    return 0;
}

//---
static const char *state_str( enum e_state st )
{
    switch ( st )
    {
    case PENDING:
        return "pending";
    case RUNNING:
        return "running";
    case WAITING:
        return "waiting";
    case DONE:
        return "done";
    case ERROR:
        return "error";
    case CANCELLED:
        return "cancelled";
    }
    assert(FALSE);  // cannot get here
    return NULL;
}

//---
// str= thread_state( lane_ud )
//
// Returns: "pending"   not started yet
//          -> "running"   started, doing its work..
//             <-> "waiting"   blocked in a receive()
//                -> "done"     finished, results are there
//                   / "error"     finished at an error, error value is there
//                   / "cancelled"   execution cancelled by M (state gone)
//
LUAG_FUNC( thread_state )
{
    struct s_lane *s= lua_touserdata(L,1);

    lua_pushstring(L,state_str(s->state));
    return 1;
}


//---
// [ "done"|"error"|"cancelled", [...] | error_val|str ]= thread_wait( lane_ud [, wait_secs=-1] )
//
// timeout:   returns nil
// done:      returns "done" + return values (0..N)
// error:     return "error" + error value
// cancelled: returns "cancelled"
//
// Note: We don't activate the error right in here, because that would
//       unnecessarily expose parts of Lanes in the error stack. Lua 'error'
//       gives us more control than C 'lua[L]_error()'.
//
LUAG_FUNC( thread_wait )
{
    struct s_lane *s= lua_touserdata(L,1);
    long wait_ms= luaL_optnumber(L,2,-1.0) * 1000;
    lua_State *L2= s->L;
    int ret;

    if (!THREAD_WAIT( &s->thread, wait_ms ))
        return 0;      // timeout: pushes none, leaves 'L2' alive

    // s->thread is very dead now, and should not be used any more.
    //
    THREAD_FREE( &s->thread );

    STACK_GROW( L, 1 );
    lua_pushstring( L, state_str(s->state) );

    if (s->state==DONE)
    {
        uint_t n= lua_gettop(L2);       // whole L2 stack
        luaG_inter_move( L2,L, n );
        ret= n+1;

    }
    else if (s->state==ERROR)
    {
        luaG_inter_move( L2,L, 1 );    // error message at [-1]
        ret= 2;

    }
    else if (s->state==CANCELLED)
    {
        ret= 1;

    }
    else
    {
        ASSERT_L( FALSE );  // unexpected state
        ret= 0;     // dummy
    }

    lua_close(L2);

    return ret;
}


/*---=== Module linkage ===---
*/

#define REG_FUNC( name ) \
    lua_pushcfunction( L, LG_##name ); \
    lua_setglobal( L, #name )

#define REG_STR2( name, val ) \
    lua_pushstring( L, val ); \
    lua_setglobal( L, #name )

LOCK_T shared_lock;

int luaopen_lanes( lua_State *L )
{

    static volatile char been_here;  // =0

    // One time initializations:
    //
    if (!been_here)
    {
        been_here= TRUE;

#if (defined PLATFORM_OSX) && (defined _UTILBINDTHREADTOCPU)
        chudInitialize();
#endif

        // Lock for "shared userdata" refcount
        //
        LOCK_INIT( &shared_lock );

        // Serialize calls to 'require' from now on
        //
        // TBD: Make in Lua level, Linux has problems with recursive Lock
#if 0
        LOCK_RECURSIVE_INIT( &require_cs );
        serialize_require( L );
#endif

        //---
        // Linux needs SCHED_RR to change thread priorities, and that is only
        // allowed for sudo'ers. SCHED_OTHER (default) has no priorities.
        // SCHED_OTHER threads are always lower priority than SCHED_RR.
        //
        // ^-- those apply to 2.6 kernel.  IF **wishful thinking** these
        //     constraints will change in the future, non-sudo priorities can
        //     be enabled also for Linux.
        //
#ifdef PLATFORM_LINUX
        sudo= geteuid()==0;     // we are root?

        // If lower priorities (-2..-1) are wanted, we need to lift the main
        // thread to SCHED_RR and 50 (medium) level. Otherwise, we're always below
        // the launched threads (even -2).
        //
#ifdef LINUX_SCHED_RR
        if (sudo)
        {
            struct sched_param sp= {0};
            sp.sched_priority= _PRIO_0;
            PT_CALL( pthread_setschedparam( pthread_self(), SCHED_RR, &sp) );
        }
#endif
#endif
        if (!init_keepers())
            luaL_error( L, "Unable to initialize" );
    }

    // Linda creator
    //
    lua_pushcfunction( L, LG_linda_new );
    lua_setglobal( L, "linda_new" );

    // metatable for threads
    //
    lua_newtable( L );
    lua_pushcfunction( L, LG_thread_gc );
    lua_setfield( L, -2, "__gc" );

    lua_pushcclosure( L, LG_thread_new, 1 );
    lua_setglobal( L, "thread_new" );

    REG_FUNC( thread_state );
    REG_FUNC( thread_wait );
    REG_FUNC( thread_cancel );

    REG_STR2( _version, VERSION );
    REG_FUNC( _yield );
    REG_FUNC( _single );

    return 0;
}
int luaopen_liblanes( lua_State *L )
{
    return luaopen_lanes( L );
}

